package com.lzy.elasticserch.canal;

import com.alibaba.otter.canal.protocol.Message;
import com.lzy.canal.model.CanalMessageEntryResultData;
import com.lzy.canal.service.CanalMessageService;
import com.lzy.elasticsearch.template.ElasticsearchServiceTemplate;
import com.lzy.elasticserch.config.CanalConsumerConfig;
import com.lzy.elasticserch.result.ResultConvert;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.ArrayUtils;
import org.elasticsearch.action.DocWriteRequest;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.slf4j.MDC;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import java.util.List;

/**
 * canal 订阅服务
 *
 */
@Slf4j
@Service
public class SubscriptionResourceService {

    private CanalConsumerConfig.CanalConsumerClient canalConsumerClient;
    @Autowired
    private CanalMessageService canalMessageService;
    @Autowired
    private ElasticsearchServiceTemplate elasticsearchClientService;
    @Autowired
    private ResultConvert resultConvert;

    /**
     * 订阅消息
     */
    @Async
    public void subscription() {
        try {
            canalConsumerClient.getCanalConnector().connect();
            canalConsumerClient.getCanalConnector().subscribe(canalConsumerClient.getCanalConsumerItemProperty().getSubscribeFilter());
            canalConsumerClient.getCanalConnector().rollback();
        } catch (Exception e) {
            log.error("同步失败", e);
            if (!canalConsumerClient.getCanalConnector().checkValid()) {
                canalConsumerClient.getCanalConnector().disconnect();
            }
            sleep();
            subscription();
        }
        while (true) {
            try {
                Message message = canalConsumerClient.getCanalConnector().getWithoutAck(1000);
                long messageId = message.getId();
                if (messageId == -1 || CollectionUtils.isEmpty(message.getEntries())) {
                    sleep();
                    continue;
                }
                MDC.put("requestId", String.valueOf(messageId));
                List<CanalMessageEntryResultData> canalMessageEntryResultData = null;
                try {
                    canalMessageEntryResultData = canalMessageService.consumerMessage(message);
                } catch (Exception e) {
                    log.error("同步失败", e);
                }
                canalConsumerClient.getCanalConnector().ack(messageId);
                if (CollectionUtils.isEmpty(canalMessageEntryResultData)) {
                    sleep();
                    continue;
                }
                DocWriteRequest<?>[] requests = resultConvert.convert(canalMessageEntryResultData);
                if (ArrayUtils.isEmpty(requests)) {
                    sleep();
                    continue;
                }
                BulkResponse response = elasticsearchClientService.bulk(new BulkRequest().add(requests));
                log.info("更新失败-->{}", response.buildFailureMessage());
            } catch (Exception e) {
                log.error("同步失败", e);
                if (!canalConsumerClient.getCanalConnector().checkValid()) {
                    canalConsumerClient.getCanalConnector().disconnect();
                }
                subscription();
            }
        }
    }

    private void sleep() {
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            log.error("sleep error:", e);
        }
    }

    public SubscriptionResourceService(CanalConsumerConfig.CanalConsumerManage canalConsumerManage) {
        this.canalConsumerClient = canalConsumerManage.getCanalConsumerResourceClient("resource");
    }
}
